package main.niit

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

import java.util.Properties


//本业务主要分析每个品牌的平均里程
// 以及里程数在5-15万公里的车辆数量
// 和这些车辆的平均售价
object mileagStatus {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.OFF) //省略运行日志

    val spark =SparkSession.builder()
      .appName("mileagStatus")
      .master("local[*]")    //本地模式运行
      .getOrCreate()
    val jdbcDF =spark.read.format("jdbc")       //从数据库中导入本地
      .option("url","jdbc:mysql://localhost:3306/cardata?useSSL=false")
      .option("driver","com.mysql.jdbc.Driver")
      .option("dbtable","data")
      .option("user","root")
      .option("password","root")
      .load()

    import spark.implicits._

    val prop =new Properties()
    prop.put("user","root")             // 数据库用户名
    prop.put("password","root")
    prop.put("driver","com.mysql.jdbc.Driver")  // 设置JDBC驱动程序的类名，用于连接MySQL数据库



    val filterDF=jdbcDF.select("汽车品牌","里程数","售卖价格").withColumn("数量",lit(1)) //lit(1)，为每一行生成一个常量值1
    val result1DF=filterDF
      .groupBy("汽车品牌")
      .agg(
        sum("数量").alias("车辆总数"),
        avg("里程数").alias("平均里程数"),
        avg("售卖价格").alias("平均售价")
      )
    //筛选符合条件的数据
    val result2DF=jdbcDF.select("汽车品牌","售卖价格").filter($"里程数" >= 50000 && $"里程数" <= 150000).withColumn("数量",lit(1))
      .groupBy("汽车品牌")
      .agg(
        sum("数量").alias("5-15万公里车辆数"),  // 计算每个品牌的5-15万公里车辆数
        avg("售卖价格").alias("平均售价1")      //以及这些车辆的平均售价
      )

    val resultDF=result1DF.join(result2DF,Seq("汽车品牌"),"left")   //将 result1DF 和 result2DF 基于 "汽车品牌" 列进行左连接

    resultDF.write.mode("append")
      .jdbc("jdbc:mysql://192.168.56.104:3306/cardata?useSSL=false","cardata. mileagStatus",prop)

  }

}
